/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/*
 * CustomInputFormat.java
 *
 * Created on Jul 9, 2010, 3:38:59 PM
 * not used
 */

package homomorphism;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
// import org.apache.commons.logging.LogFactory;
// import org.apache.commons.logging.Log;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileRecordReader;

/**
 *
 * @author takeshi
 */
public class CustomInputFormat
        extends SequenceFileInputFormat<LongWritable,LongWritable>
        implements InputFormat<LongWritable,LongWritable> {

    public static class CustomInputFormatRecordReader implements RecordReader<LongWritable,LongWritable> {
        private JobConf job;
        private FileSplit split;
        private String separator;
        private SequenceFileRecordReader in;
        private Counters.Counter inputRecordCounter;

        public CustomInputFormatRecordReader(JobConf job, FileSplit split) {
            this.job = job;
            this.split = split;
            try {

                this.in = new SequenceFileRecordReader(job, split);
                // TODO Initialize the RecordReader
                // TODO Initialize the RecordReader
            } catch (IOException ex) {
                Logger.getLogger(CustomInputFormat.class.getName()).log(Level.SEVERE, null, ex);
            }
        }



        @Override
        public LongWritable createKey() {
           long num = -1;
            try {
                num = this.getPos();
            } catch (IOException ex) {
                Logger.getLogger(CustomInputFormat.class.getName()).log(Level.SEVERE, null, ex);
            }
            return new LongWritable(num);
        }

        @Override
        public LongWritable createValue() {
            return  (LongWritable)in.createValue();
           // return new LongWritable();
        }

        @Override
        public synchronized boolean next(LongWritable key, LongWritable value)
                throws IOException {
            // TODO Parse a record
            return in.next(key, value);
        }

        @Override
        public long getPos() throws IOException {
            // TODO Compute the position in the split, and return it.
            return in.getPos();
        }

        @Override
        public float getProgress() throws IOException {
            return in.getProgress();
        }

        @Override
        public void close() throws IOException {
             in.close();
        }
    }

    // do not split the file .... means I can only  use just 1 mapper ...
    @Override
    protected boolean isSplitable(FileSystem fs , Path filename){
        return false;
    }

    public RecordReader<LongWritable, LongWritable> getRecordReader(InputSplit split, JobConf job, Reporter reporter)
                        throws IOException {
        reporter.setStatus(split.toString());
        return new CustomInputFormatRecordReader(job, (FileSplit)split);
    }
}